package xj.toolkit.netty.client;

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import xj.toolkit.base.Sender;
import xj.toolkit.netty.initializer.TcpClientInitializer;
import xj.toolkit.netty.transport.TransportUtils;
import xj.toolkit.netty.transport.endpoint.ClientEndpoint;
import xj.toolkit.netty.transport.endpoint.Endpoint;

/**
 */
public abstract class TCPConnector implements Sender {
	private static final Logger logger = LoggerFactory.getLogger(TCPConnector.class);

	private ScheduledExecutorService exec = null;

	private String name = "connector";
	private String destIp = null;
	private int destPort = -1;
	private String localAddress = "0.0.0.0";
	private int localPort;
	private Channel channel = null;
	private TcpClientInitializer tcpClientInitializer;
	private Bootstrap bootstrap = new Bootstrap();
	private NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
	private long retryTimeout = 1000;
	private long waitTimeout = 1000;

	@ChannelHandler.Sharable
	private class IOHandler extends ChannelInboundHandlerAdapter {

		@Override
		@SuppressWarnings("unchecked")
		public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
			Endpoint endpoint = TransportUtils.getEndpointOfSession(ctx);
			if (null != endpoint) {
				endpoint.messageReceived(ctx, msg);
			} else {
				if (logger.isWarnEnabled())
					logger.warn("{} missing endpoint, ignore incoming msg:{}", getDesc(), msg);
			}
		}

		@Override
		public void channelActive(ChannelHandlerContext ctx) throws Exception {
			logger.info("channelActive:{}", ctx.channel());
			try {
				channel = ctx.channel();
				// create endpoint
				ClientEndpoint endpoint = createEndpoint();
				if (null != endpoint) {
					TransportUtils.attachEndpointToSession(ctx, endpoint);
					endpoint.setChannel(channel);
				}
			} catch (Exception ex) {
				logger.warn(" createEndpoint:", ex);
				channel = null;
				ctx.channel().close();
			}
		}

		@Override
		public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
			Endpoint endpoint = TransportUtils.getEndpointOfSession(ctx);
			if (null != endpoint) {
				endpoint.stop();
			}
			if (!exec.isShutdown()) {
				exec.submit(new Runnable() {

					public void run() {
						onChannelDisconnected(ctx.channel());
					}
				});
			}
		}

		@Override
		public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
			logger.error(getDesc() + " TCPConnector:", cause);
		}
	}

	private void doConnect() {
		if (null == destIp || destIp.equals("")) {
			logger.warn(getDesc() + " destIp is null, disable this connector.");
			return;
		}

		this.tcpClientInitializer.addOtherHandler(new IOHandler());

		ChannelFuture future = bootstrap.connect();

		future.addListener(new ChannelFutureListener() {

			public void operationComplete(final ChannelFuture future) throws Exception {
				exec.submit(new Runnable() {
					public void run() {
						onConnectComplete(future);
					}
				});
			}
		});
	}

	private void onConnectComplete(ChannelFuture future) {
		if (!future.isSuccess()) {
			if (logger.isInfoEnabled()) {
				logger.info(getDesc() + " connect [" + this.destIp + ":" + this.destPort + "] failed, retry...");
			}
			exec.schedule(new Runnable() {

				public void run() {
					doConnect();
				}
			}, retryTimeout, TimeUnit.MILLISECONDS);
		} else {
			// 连接成功
			InetSocketAddress address = (InetSocketAddress) future.channel().localAddress();
			localAddress = address.getAddress().getHostAddress();
			localPort = address.getPort();
		}
	}

	private void onChannelDisconnected(Channel channel) {
		if (logger.isInfoEnabled()) {
			logger.info(getDesc() + " channel : " + channel + "closed, retry connect...");
		}
		exec.schedule(new Runnable() {

			public void run() {
				doConnect();
			}
		}, retryTimeout, TimeUnit.MILLISECONDS);
	}

	public TCPConnector() {
	}

	public TCPConnector(String name) {
		this.name = name;
	}

	public String getDestIp() {
		return destIp;
	}

	public void setDestIp(String destIp) {
		this.destIp = destIp;
	}

	public int getDestPort() {
		return destPort;
	}

	public void setDestPort(int destPort) {
		this.destPort = destPort;
	}

	public TcpClientInitializer getTcpClientInitializer() {
		return tcpClientInitializer;
	}

	public void setTcpClientInitializer(TcpClientInitializer tcpClientInitializer) {
		this.tcpClientInitializer = tcpClientInitializer;
	}

	public void start() {
		exec = Executors.newSingleThreadScheduledExecutor();
		bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class).localAddress(localAddress, localPort)
				.remoteAddress(destIp, destPort);
		bootstrap.handler(tcpClientInitializer);
		doConnect();
	}

	public void stop() {
		if (this.exec != null) {
			this.exec.shutdownNow();
		}
		this.eventLoopGroup.shutdownGracefully();
		if (this.channel != null) {
			this.channel.close();
		}
	}

	private String getDesc() {
		return "[" + name + "]";
	}

	public long getRetryTimeout() {
		return retryTimeout;
	}

	public void setRetryTimeout(long retryTimeout) {
		this.retryTimeout = retryTimeout;
	}

	public String getLocalAddress() {
		return localAddress;
	}

	public void setLocalAddress(String localAddress) {
		this.localAddress = localAddress;
	}

	public int getLocalPort() {
		return localPort;
	}

	public void setLocalPort(int localPort) {
		this.localPort = localPort;
	}

	public long getWaitTimeout() {
		return waitTimeout;
	}
	
	public boolean isActive() {
		return channel != null && channel.isActive();
	}

	protected abstract ClientEndpoint createEndpoint();
	
	@Override
	public void send(Object target) {
		Preconditions.checkArgument(target != null);
		if (channel != null && channel.isActive() && channel.isWritable()) {
			channel.writeAndFlush(target);
			return;
		}

		throw new IllegalStateException("Channel is NULL or INACTIVE or CANNOT WRITE.");
	}
	

	@Override
	public void send(Object req, Object response) {
		send(response);
	}

}
